package com.cetccloud.ap.controller;

import cn.hutool.core.map.MapUtil;
import com.alibaba.fastjson.JSON;
import com.cetccloud.ap.controller.model.ApiResult;
import com.cetccloud.ap.controller.model.req.MockReq;
import com.cetccloud.ap.util.KafkaKeyUtil;
import com.fasterxml.jackson.annotation.JsonInclude;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.List;

/**
 * @ClassName MockController
 * @Description TODO
 * @Author superz
 * @Date 2021/4/19 16:58
 * @Version 1.0
 **/
@Slf4j
@RestController
@RequestMapping("/mock")
public class MockController {

    @PostMapping("/object")
    public Mono<ApiResult> object(@RequestBody MockReq mockReq,
                                  ServerHttpRequest request, ServerHttpResponse response) {
        log.info("/mock/object: {}", JSON.toJSONString(mockReq));
        return Mono.just(ApiResult.success().build(MapUtil.of(3, "func3")))
                .map(apiResult -> {
                    int i = 1 / 0;
                    return apiResult;
                })
                .onErrorResume(throwable -> Mono.just(ApiResult.error(throwable.getMessage())));
    }

    @GetMapping("/array/{size}")
    public Mono<Object> array(@PathVariable("size") int size) {
        log.info("/mock/array/{}", size);
        List list = new ArrayList<>();
        for (int i = 1; i <= size; i++) {
            list.add(MapUtil.of(i,"func"+i));
        }
        return Mono.just(list);
    }

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @PostMapping("/kafka")
    public Mono<Object> kafka(@RequestBody MockReq mockReq,
                              ServerHttpRequest request, ServerHttpResponse response) {
        log.info("/mock/object: {}", JSON.toJSONString(mockReq));
        ProducerRecord<String, String> producerRecord =
                new ProducerRecord<>("topic.proxy.transfer.topic1", KafkaKeyUtil.zoneByKey("zoneId"), "message content");
        kafkaTemplate.send(producerRecord);
        return Mono.empty();
    }
}
